package io.github.eric.urpc.core.conn;

import io.github.eric.urpc.core.config.AbstractContext;
import io.github.eric.urpc.core.exception.UrpcMsgSendException;
import io.github.eric.urpc.core.rpc.Message;
import io.github.eric.urpc.core.rpc.Request;
import io.github.eric.urpc.core.rpc.Response;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.util.AttributeKey;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


public class Connection {
	public static final AttributeKey<Connection> TCP = AttributeKey.valueOf("tcp");
	private static final byte INIT_IN_PROGRESS = 0;
	private static final byte INIT_COMPLETED = 1;
	private static final byte INIT_FAILED = -1;

	private final String id;
	private final Channel channel;
	private final AbstractContext context;
	private final Map<String, Object> attributes = new ConcurrentHashMap<>();
	/**
	 * requestId -> reqResPair
	 */
	private final Map<String, PendingRequest> awaitingResponseRequests = new ConcurrentHashMap<>();

	private byte initStatus;

	public Connection(Channel channel, AbstractContext context) {
		this.channel = channel;
		this.id = channel.id().asShortText();
		this.context = context;
	}

	public void sendAsync(Message message) {
		if (Request.class.isAssignableFrom(message.getClass())) {
			throw new UrpcMsgSendException("[urpc] async send method can't send Request msg, please use request method");
		}
		sendAsync0(message);
	}

	private void sendAsync0(Object data) {
		assertConnectionWriteable();
		channel.writeAndFlush(data);
	}

	private void assertConnectionWriteable() {
		if (!channel.isOpen()) {
			throw new UrpcMsgSendException("[urpc] send msg failed, channel is not open");
		}
		if (!channel.isActive()) {
			throw new UrpcMsgSendException("[urpc] send msg failed, channel is not active");
		}
		if (!channel.isWritable()) {
			throw new UrpcMsgSendException("[urpc] send msg failed, channel is not writable");
		}
	}

	public <T extends Response> T sendSync(Request request) {
		return sendSync(request, context.getReadTimeout()*1000L);
	}

	@SuppressWarnings("unchecked")
	public <T extends Response> T sendSync(Request request, long readTimeout) {
		if (request == null) {
			throw new UrpcMsgSendException("[urpc] request is not present");
		}
		PendingRequest pendingRequest = new PendingRequest(request);
		try {
			pendingRequest.getLock().lock();
			awaitingResponseRequests.put(request.getRequestId(), pendingRequest);
			sendAsync0(request);
			final long deadline = System.currentTimeMillis() + readTimeout;
			long remaining = readTimeout;
			while (remaining > 0) {
				pendingRequest.getCondition().await(remaining, TimeUnit.MILLISECONDS);
				if (pendingRequest.getResponse() != null) {
					break;
				}
				remaining = deadline - System.currentTimeMillis();
			}
			Response response = pendingRequest.getResponse();
			if (response == null) {
				throw new UrpcMsgSendException("[urpc] sync send request read timeout");
			}
			return (T) response;
		} catch (Exception e) {
			if (e instanceof InterruptedException) {
				Thread.currentThread().interrupt();
			}
			awaitingResponseRequests.remove(request.getRequestId());
			throw new UrpcMsgSendException("[urpc] sync send request failed", e);
		} finally {
			pendingRequest.getLock().unlock();
		}
	}

	void onResponseArrived(Response response) {
		PendingRequest pendingRequest = awaitingResponseRequests.remove(response.getRequestId());
		if (pendingRequest == null) {
			throw new UrpcMsgSendException(String.format("[urpc] notify response failed, can not find pending request of requestId %s", response.getRequestId()));
		}
		try {
			pendingRequest.getLock().lock();
			pendingRequest.setResponse(response);
			pendingRequest.getCondition().signalAll();
		} catch (Exception e) {
			throw new UrpcMsgSendException("[urpc] notify response failed", e);
		} finally {
			pendingRequest.getLock().unlock();
		}
	}

	public void onConnected() {
		for (ConnectionListener listener : context.getConnectionListeners()) {
			listener.onConnected(this);
		}
	}

	public void onDisconnected() {
		for (ConnectionListener listener : context.getConnectionListeners()) {
			listener.onDisConnected(this);
		}
	}

	public void onAccepted() {
		for (ConnectionListener listener : context.getConnectionListeners()) {
			listener.onAccepted(this);
		}
	}

	public boolean isConnected() {
		return channel != null && channel.isOpen() && channel.isRegistered() && channel.isActive();
	}

	public ChannelFuture close() {
		return channel.close().syncUninterruptibly();
	}

	public ChannelFuture closeAsync() {
		return channel.close();
	}

	public Object getAttribute(String name) {
		return attributes.get(name);
	}

	public void putAttribute(String name, Object object) {
		attributes.put(name, object);
	}

	public String getId() {
		return id;
	}

	public InetSocketAddress getLocalAddress() {
		return (InetSocketAddress) channel.localAddress();
	}

	public InetSocketAddress getRemoteAddress() {
		return (InetSocketAddress) channel.remoteAddress();
	}

	public void setInitStatusCompleted() {
		if (!isInitInProgress()) {
			return;
		}
		initStatus = INIT_COMPLETED;
	}

	public void setInitStatusFailed() {
		if (!isInitInProgress()) {
			return;
		}
		initStatus = INIT_FAILED;
	}

	public boolean isInitInProgress() {
		return initStatus == INIT_IN_PROGRESS;
	}

	public boolean isInitCompleted() {
		return initStatus == INIT_COMPLETED;
	}

	public boolean isInitFailed() {
		return initStatus == INIT_FAILED;
	}

	static class PendingRequest {
		private final Request request;
		private Response response;
		private final ReentrantLock lock;
		private final Condition condition;

		PendingRequest(Request request) {
			this.request = request;
			this.lock = new ReentrantLock();
			this.condition = lock.newCondition();
		}

		public Request getRequest() {
			return request;
		}

		public Response getResponse() {
			return response;
		}

		public void setResponse(Response response) {
			this.response = response;
		}
		public ReentrantLock getLock() {
			return lock;
		}

		public Condition getCondition() {
			return condition;
		}
	}

}
